package logic

import (
	"context"
	"encoding/json"
	"io"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"paper-pro/app/paperfile/model"
	"paper-pro/app/paperfile/rpc/internal/svc"
	"paper-pro/app/paperfile/rpc/pb/paperfile"
	"paper-pro/pkg/xerr"

	"github.com/minio/minio-go/v7"
	"github.com/pkg/errors"
	"github.com/zeromicro/go-zero/core/logx"
	"go.opentelemetry.io/otel/trace"
)

var ErrFileExisted = xerr.NewErrCodeMsg(xerr.SERVER_COMMON_ERROR, "文件已存在")
var ErrRecvFileMetaFailed = xerr.NewErrCodeMsg(xerr.SERVER_COMMON_ERROR, "获取文件元数据信息失败")

type UploadLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewUploadLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UploadLogic {
	return &UploadLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

func (l *UploadLogic) pushKqParseTask(paperMeta *model.Paper) error {
	payload, err := json.Marshal(CopyKqPaperMeta(paperMeta))
	if err != nil {
		return err
	}
	kqTaskMsg := KqPdfTaskInfo{
		TaskType:     PDF_PARSE,
		TaskPlayload: string(payload),
	}
	msgBytes, err := json.Marshal(kqTaskMsg)
	if err != nil {
		return err
	}
	return l.svcCtx.KqPusherClient.PushWithKey(l.ctx, paperMeta.ID.Hex(), string(msgBytes)) // 保证相同文件排在同一分区，从而保证任务的顺序
}

func (l *UploadLogic) Upload(stream paperfile.PaperFile_UploadServer) (err error) {
	// todo: 流的拦截器得另外实现，后续再搞吧
	defer func() {
		if err != nil {
			l.Errorf("【RPC-SRV-ERR】Upload error: %+v", err)
		}
	}()

	// 1. 创建临时文件
	tmpFile, err := os.CreateTemp("", "upload-*.tmp")
	if err != nil {
		return errors.Wrapf(err, "create temp file failed")
	}
	defer os.Remove(tmpFile.Name())
	defer tmpFile.Close()

	// 2. 流式写入临时文件
	paperMeta, err := l.recvPaperFile(stream, tmpFile)
	if err != nil {
		return err
	}
	defer func() {
		// 若中途有任何失败，删除插入的元数据
		if paperMeta.Status == model.StatusFailed {
			bgCtx, cancel := newCompensateOpCtx(l.ctx, 3)
			defer cancel()
			l.svcCtx.PaperModel.Delete(bgCtx, paperMeta.ID.Hex())
		}
	}()

	// 3. 上传MinIO
	objectName := generateObjectName(paperMeta.UserId, paperMeta.FileName)
	if _, err := tmpFile.Seek(0, 0); err != nil {
		return errors.Wrapf(err, "reset file pointer failed")
	}

	uploadInfo, err := l.svcCtx.MinioClient.PutObject(l.ctx,
		paperMeta.MinIOBucket,
		paperMeta.MinIOObject,
		tmpFile,
		-1,
		minio.PutObjectOptions{ContentType: paperMeta.ContentType},
	)
	if err != nil {
		paperMeta.Status = model.StatusFailed
		return errors.Wrapf(err, "minio upload failed")
	}

	// 4. 写入MongoDB(第二次更新文件元数据状态)
	paperMeta.Status = model.StatusUploaded
	paperMeta.FileSize = uploadInfo.Size
	paperMeta.MinIOBucket = uploadInfo.Bucket
	paperMeta.MinIOObject = objectName
	paperMeta.ETag = uploadInfo.ETag
	paperMeta.ParseStatus = model.ParsePending
	if _, err := l.svcCtx.PaperModel.Update(l.ctx, paperMeta); err != nil {
		// 补偿：删除已上传的MinIO对象
		bgCtx, cancel := newCompensateOpCtx(l.ctx, 3)
		defer cancel()
		if delErr := l.svcCtx.MinioClient.RemoveObject(bgCtx,
			l.svcCtx.Config.MinioConf.Bucket,
			objectName,
			minio.RemoveObjectOptions{},
		); delErr != nil {
			l.Errorf("compensation failed: %v", delErr)
		}
		paperMeta.Status = model.StatusFailed
		return errors.Wrapf(err, "mongo insert failed")
	}

	// 5. 发布解析任务到消息队列
	if err := l.pushKqParseTask(paperMeta); err != nil {
		l.Errorf("【RPC-SRV-ERR】Upload error: push parse task to kafka error: %+v", err)
	}

	// 6. 返回结果
	return stream.SendAndClose(&paperfile.PaperFileInfo{
		Meta: &paperfile.PaperFileMeta{
			Id:          paperMeta.ID.Hex(),
			UpdateAt:    paperMeta.UpdateAt.Unix(),
			CreateAt:    paperMeta.CreateAt.Unix(),
			FileName:    paperMeta.FileName,
			FileSize:    paperMeta.FileSize,
			ContentType: paperMeta.ContentType,
			Status:      paperMeta.Status,
			ParseStatus: paperMeta.ParseStatus,
		},
		MinIOInfo: &paperfile.MinIOInfo{
			MinIOBucket:  paperMeta.MinIOBucket,
			MinIOObject:  paperMeta.MinIOObject,
			ETag:         paperMeta.ETag,
			StorageClass: "", // 未使用
		},
		VectorDBInfo: &paperfile.VectorDBInfo{
			ParseStatus: paperMeta.ParseStatus,
		},
	})
}

// 提供补偿操作独立上下文，保证因超时引起的失败补偿操作不受原ctx超时的影响
func newCompensateOpCtx(originalCtx context.Context, seconds time.Duration) (context.Context, context.CancelFunc) {
	span := trace.SpanFromContext(originalCtx)
	newCtx, cancel := context.WithTimeout(context.Background(), seconds*time.Second)
	return trace.ContextWithSpan(newCtx, span), cancel
}

func (l *UploadLogic) recvPaperFile(stream paperfile.PaperFile_UploadServer, tmpFile *os.File) (paperMeta *model.Paper, err error) {
	var wg sync.WaitGroup
	var checkErr error = ErrRecvFileMetaFailed
	defer func() {
		wg.Wait()
		if err != nil && checkErr == nil { // 代表元数据已插入mongo数据库，需删除
			if paperMeta != nil && !paperMeta.ID.IsZero() {
				l.svcCtx.PaperModel.Delete(l.ctx, paperMeta.ID.Hex())
				paperMeta = nil
			}
		}
		if checkErr != nil {
			paperMeta = nil
			err = checkErr
		}
	}()

	paperMeta = &model.Paper{}
	paperMeta.Status = model.StatusStarting
	var isError atomic.Bool
	for !isError.Load() {
		chunk, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			isError.Store(true)
			return paperMeta, errors.Wrapf(err, "stream receive failed")
		}

		if _, err := tmpFile.Write(chunk.Content); err != nil {
			isError.Store(true)
			return paperMeta, errors.Wrapf(err, "write temp file failed")
		}

		// 只解析第一次收到的元数据
		if paperMeta.Status == model.StatusStarting && chunk.UploadReqMeta != nil {
			checkErr = nil
			paperMeta.FileName = chunk.UploadReqMeta.FileName
			paperMeta.FileSize = chunk.UploadReqMeta.FileSize
			paperMeta.ContentType = chunk.UploadReqMeta.ContentType
			paperMeta.UserId = chunk.UploadReqMeta.UserId
			paperMeta.Status = model.StatusProcessing
			paperMeta.MinIOBucket = l.svcCtx.Config.MinioConf.Bucket
			paperMeta.MinIOObject = generateObjectName(paperMeta.UserId, paperMeta.FileName)
			paperMeta.ParseStatus = model.ParsePending
			// 异步检查MinIO中是否已存在该文件，同时第一次插入元数据
			wg.Add(1)
			go func() {
				defer wg.Done()
				if isError.Load() {
					return
				}
				exists, err := l.checkObjectExists(paperMeta)
				if err != nil {
					checkErr = errors.Wrap(err, "check object existence failed")
					isError.Store(true)
					return
				}
				if exists {
					checkErr = ErrFileExisted
					isError.Store(true)
					return
				}
				if isError.Load() {
					return
				}
				if err := l.svcCtx.PaperModel.Insert(l.ctx, paperMeta); err != nil { // 注意！在Insert时，paperMeta在其他位置是只读的
					checkErr = errors.Wrap(err, "mongo insert file meta failed")
					isError.Store(true)
					return
				}
			}()
		}
	}

	return paperMeta, nil
}

func generateObjectName(userId int64, fileName string) string {
	dirName := strconv.FormatInt(userId, 10)
	fileName = strings.ReplaceAll(fileName, "/", "_")
	fileName = strings.ReplaceAll(fileName, "..", "") // 防止路径穿越
	return filepath.Join(dirName, fileName)
}

func (l *UploadLogic) checkObjectExists(fileMeta *model.Paper) (bool, error) {
	_, err := l.svcCtx.MinioClient.StatObject(
		l.ctx,
		fileMeta.MinIOBucket,
		fileMeta.MinIOObject,
		minio.StatObjectOptions{},
	)
	if err != nil {
		if errResp, ok := err.(minio.ErrorResponse); ok {
			if errResp.Code == "NoSuchKey" {
				return false, nil
			}
		}
		return false, err
	}
	return true, nil
}
